package com.xzc.apitest.udftest

import com.xzc.apitest.source.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.scala._
import org.apache.flink.table.functions.TableAggregateFunction
import org.apache.flink.types.Row
import org.apache.flink.util.Collector

object TableAggregateFunctionTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //1.11之后默认就是这个，之前 OlePlanner
    val settings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    //创建表环境
    val tableEnv = StreamTableEnvironment.create(env, settings)

    val inputStream = env.readTextFile("D:\\git\\learning_flink\\_01_试用\\src\\main\\resources\\sensor.txt")

    //将流转换为样例类形式
    val dataStream = inputStream
      .map(data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }).assignTimestampsAndWatermarks(
      //在转换为表之前就将流的watermark指定，从而在表里面可以直接使用
      new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(3)) {
        override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000L
      }
    )

    //建立动态表，最后的字段是flink给的时间特性
    //    val sensorTable = tableEnv.fromDataStream(dataStream,
    //      'id, 'temperature, 'timestamp, 'pt.proctime)
    //这里的rowtime就是上面流里面定义的watermark，覆盖了原有的timestamp
    val sensorTable = tableEnv.fromDataStream(dataStream,
      'id, 'temperature, 'timestamp.rowtime as 'ts)

    //table api
    val top2TempFunc = new Top2TempFunc
    val resultTable = sensorTable
      .groupBy('id)
      .flatAggregate(top2TempFunc('temperature) as('temp, 'rank))
      .select('id, 'temp, 'rank)

    //sql api 不太好用，这里没有演示

    resultTable.toRetractStream[Row].print()

    env.execute("table aggregate function test")
  }
}

//定义一个类，专门用于表示表聚合的状态
//因为元组是无法更新其值的，所以这里要对象来代替，每次来新数据就更新
class Top2TempAcc {
  //因为是极值，所以用静态变量没事，但如果是分布式多管道执行，这里是不是不行
  var highestTemp: Double = Double.MinValue
  var secondHighestTemp: Double = Double.MinValue
}

//自定义表聚合函数，提取所有温度值中最高的两个温度
//输出(temp, rank)
class Top2TempFunc extends TableAggregateFunction[(Double, Int), Top2TempAcc] {
  override def createAccumulator(): Top2TempAcc = new Top2TempAcc

  //实现计算聚合结果的函数，没有重写限制，但方法名要相同
  def accumulate(acc: Top2TempAcc, temp: Double): Unit = {
    //判断当前温度值，是否比状态中的值大
    if (temp > acc.highestTemp) {
      //如果比最高温度还高，排在第一，原来的第一顺到第二位
      acc.highestTemp = temp
    } else if (temp > acc.secondHighestTemp) {
      //如果在最高和第2高之间，那么直接替换第2高温度
      acc.secondHighestTemp = temp
    }
  }

  //实现一个输出结果的方法，最终处理完表中所有数据时调用
  def emitValue(acc: Top2TempAcc, out: Collector[(Double, Int)]): Unit = {
    out.collect((acc.highestTemp, 1))
    out.collect((acc.secondHighestTemp, 2))
  }
}
